package net.bwie.realtime.jtp.dwd.log.jjb;


import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.dispatcher.runner.DispatcherRunner;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.jtp.dwd.log.jjb
 * @Author: zhangleqing
 * @CreateTime: 2025-08-17  19:32
 * @Description: TODO
 * @Version: 1.0
 */
public class Text {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "topic-log");

        // 转换+清洗+处理
        DataStream<String> pageStream = pageSteam(kafkaStream);

        // 写入+展示
        //KafkaUtil.producerKafka(pageStream, "dwd-traffic-page-log");
        pageStream.print("清理后的展示==>");

        env.execute("Text");
    }

    private static DataStream<String> pageSteam(DataStream<String> stream) {
        // 数据清洗
        DataStream<String> t1 = tt1(stream);
        // 新老用户修复
        DataStream<String> t2 = tt2(t1);
        // 分流
        DataStream<String> t3 = tt3(t2);

        return t3;
    }


    // 分流
    private static DataStream<String> tt3(DataStream<String> stream) {
        // 设置侧流
        final OutputTag<String> errorStream = new OutputTag<String>("error-log") {};
        final OutputTag<String> startStream = new OutputTag<String>("start-log") {};
        final OutputTag<String> displayStream = new OutputTag<String>("display-log") {};
        final OutputTag<String> actionStream = new OutputTag<String>("action-log") {};

        // 使用继承完成
        SingleOutputStreamOperator<String> steam = stream.process(
                new Text_02(errorStream, startStream, displayStream, actionStream)
        );

        // 就是获取每个侧流的数据，然后写入kafka没有技术含量
        SideOutputDataStream<String> error = steam.getSideOutput(errorStream);
        KafkaUtil.producerKafka(error, "dwd-traffic-error-log");
        DataStream<String> start = steam.getSideOutput(startStream);
        KafkaUtil.producerKafka(start, "dwd-traffic-start-log");
        DataStream<String> display = steam.getSideOutput(displayStream);
        KafkaUtil.producerKafka(display, "dwd-traffic-display-log");
        DataStream<String> action = steam.getSideOutput(actionStream);
        KafkaUtil.producerKafka(action, "dwd-traffic-action-log");
        return steam;
    }

    // 新老用户修复
    private static DataStream<String> tt2(DataStream<String> stream) {
        // 获取mid并分组
        KeyedStream<String, String> midStream = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String s) throws Exception {
                return JSON.parseObject(s).getJSONObject("common").getString("mid");
            }
        });

        // 使用继承完成
        DataStream<String> steam = midStream.process(new Text_01());
        return steam;
    }

    // 数据清洗
    private static DataStream<String> tt1(DataStream<String> stream) {
        // 创建一个容器，存储脏数据
        final OutputTag<String> dirtyTag = new OutputTag<>("dirty-log");

        // 过滤非json数据
        SingleOutputStreamOperator<String> steam = stream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value,
                                       Context ctx,
                                       Collector<String> out) throws Exception {
                try {
                    JSON.parseObject(value);
                    out.collect(value);
                } catch (Exception e) {
                    ctx.output(dirtyTag, value);
                }
            }
        });

        // 把脏数据写入kafka，供后续分析使用
        SideOutputDataStream<String> dirty = steam.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirty, "dwd-traffic-dirty-log");

        return steam;
    }

}
